package com.zhao.demo.unbound.demo04_transformation.sample03_split


import com.zhao.demo.Raytek
import org.apache.flink.streaming.api.scala.{SplitStream, StreamExecutionEnvironment}

/**
 * Description: 使用DataStream中transformation算子将流分成不同特质的数据<br/>
 * Copyright (c) ，2020 ， 赵 <br/>
 * This program is protected by copyright laws. <br/>
 * Date： 2020/12/8 16:20
 *
 * @author 柒柒
 * @version : 1.0
 */

object SplitStreamDemo {
  def main(args: Array[String]): Unit = {
    //1.环境
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    //2.实时读取流数据,计算,并显示结果
    import org.apache.flink.api.scala._

    val splitStream: SplitStream[Raytek] = env.socketTextStream("node01", 7775)
      .map(perInfo => {
        val arr = perInfo.split(",")
        val id = arr(0).trim
        val temperature = arr(1).trim.toDouble
        val name = arr(2).trim
        val timestamp = arr(3).trim.toLong
        val location = arr(4).trim
        Raytek(id, temperature, name, timestamp, location)
      }).split((raytect: Raytek) => if (raytect.temperature >= 36.3 && raytect.temperature <= 37.2) Seq("正常") else Seq("异常"))

//    val splitStream: SplitStream[Raytek] = env.socketTextStream("node01", 7775)
//      .map(perInfo => {
//        val arr = perInfo.split(",")
//        val id = arr(0).trim
//        val temperature = arr(1).trim.toDouble
//        val name = arr(2).trim
//        val timestamp = arr(3).trim.toLong
//        val location = arr(4).trim
//        Raytek(id, temperature, name, timestamp, location)
//      }).split((raytect: Raytek) => if (raytect.temperature >= 36.3 && raytect.temperature <= 37.2) Seq("正常") else Seq("异常"))

    //针对不同特征的旅客,进行不同的处理
    //a)从流中取出所有体温正常的旅客信息,继续进行处理(显示)
    splitStream.select("正常")
      .print("体温正常的旅客信息->")

    //b)从流中取出所有体温异常的旅客信息,进行处理
    splitStream.select("异常")
      .print("体温异常的旅客信息->")

    //3.启动
    env.execute()
  }
}














